package com.quarkus.aggregator.utils;

import io.vertx.core.json.JsonObject;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

import java.security.KeyPair;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @version v1.0
 * @Description: TODO(一句话描述该类的功能)
 * @Author: TongRui乀
 * @Date: 2021/3/6 11:11 上午
 */
public class QStream  {

    private KStream stream;

    private StreamsBuilder builder;

    public QStream(){
        builder  = new StreamsBuilder();
    }

    public QStream(String topic){
        builder  = new StreamsBuilder();
        this.stream(topic);
    }

    /**
     * @Description:
     * @param @topic Kafka topic
     * @return @return KStream
     * @author TongRui乀
     * @throws
     * @date 2021/3/6 11:32 上午
     */
    public QStream stream(String topic){
        this.stream = builder.stream(topic);
        return this;
    }

    public Select select(){
        return new Select(stream);
    }

    public Save save(){
        return new Save();
    }

    class Select <K, V>{

        private KStream<K,V> stream;

        private Map<String, String> types = new HashMap<>();

        public Select(KStream<K,V> stream){
            this.stream = stream;
            // 处理source
            handleSource();
        }

        /**
         * @Description: 处理消息
         * @throws
         * @date 2021/3/6 1:56 下午
         */
        public void handleSource(){

            stream.filter((id, message) -> Objects.nonNull(message))
                    .map((id, message)->{
                var body = new JsonObject(message.toString());
                return KeyValue.pair(id, body);
            }).foreach((id, body)->{
                System.out.println(id);
                System.out.println(body.toString());

                body.containsKey("after");
            });
        }

        public Select<K,V> string(String column){
            types.put(column, "String");
            return this;
        }



    }

    class Save{



    }


    public static void main(String[] args) {

        QStream qStream = new QStream("test");

        qStream.select();

    }
}
